package com.xiaour.spring.boot.rocketmq.controller;
import com.xiaour.spring.boot.rocketmq.consumer.LogisticsConsumer;
import com.xiaour.spring.boot.rocketmq.consumer.friendConsumer;
import com.xiaour.spring.boot.rocketmq.consumer.notifyConsumer;
import com.xiaour.spring.boot.rocketmq.producer.LogisticsInfor;
import com.xiaour.spring.boot.rocketmq.producer.Producer;
import com.xiaour.spring.boot.rocketmq.producer.friendProducer;
import com.xiaour.spring.boot.rocketmq.producer.notifyinfor;

import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.PullResult;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.io.UnsupportedEncodingException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;

@RestController
public class RocketController {
	    @Autowired
	    private Producer producer;
	    @Autowired
	    private friendProducer friendproducer;
	    @Autowired
	    private LogisticsInfor logisticsproducer;
	    @Autowired
	    private friendConsumer friendconsumer;
	    @Autowired
	    private LogisticsConsumer logisticsconsumer;
	    @Autowired
	    private notifyConsumer notifyconsumer;
	    @Autowired
	    private notifyinfor notifyproducer;
	    /**
	     * 消费者
	     */
	    @Value("${apache.rocketmq.consumer.pushConsumer}")
	    private String pushConsumer;

	    /**
	     * NameServer 地址
	     */
	    @Value("${apache.rocketmq.namesrvAddr}")
	    private String namesrvAddr;

	    @RequestMapping("/push")
	    public String pushMsg(String msg){
	        try {
	        	System.out.println("短信内容：" + msg);
	        	if(msg.isEmpty())
	        	{
	        		return "{\"result\": \"ERROR\", \"msg\":\"消息内容不能为空\"}";
	        	}
	        	String send = producer.send("BenchmarkTest","push1",msg);
	        } catch (InterruptedException e) {
	            e.printStackTrace();
	        } catch (RemotingException e) {
	            e.printStackTrace();
	        } catch (MQClientException e) {
	            e.printStackTrace();
	        } catch (MQBrokerException e) {
	            e.printStackTrace();
	        } catch (UnsupportedEncodingException e) {
	            e.printStackTrace();
	        }
	        return "ERROR";
	    }
	    @RequestMapping("/pushnotify")
	    public String pushnotify(String msg){
	        try {
	        	System.out.println("短信内容：" + msg);
	        	if(msg.isEmpty())
	        	{
	        		return "{\"result\": \"ERROR\", \"msg\":\"消息内容不能为空\"}";
	        	}
	            return notifyproducer.send("notifyinfor","notif",msg);
	        } catch (InterruptedException e) {
	            e.printStackTrace();
	        } catch (RemotingException e) {
	            e.printStackTrace();
	        } catch (MQClientException e) {
	            e.printStackTrace();
	        } catch (MQBrokerException e) {
	            e.printStackTrace();
	        } catch (UnsupportedEncodingException e) {
	            e.printStackTrace();
	        }
	        return "ERROR";
	    }
	    @RequestMapping("/PushFriendMsg")
	    public String PushFriendMsg(String userid, String msg){
	        try {
	        	System.out.println("短信内容：" + msg);
	        	System.out.println("userid：" + userid);
	        	if(msg.isEmpty())
	        	{
	        		return "{\"result\": \"ERROR\", \"msg\":\"消息内容不能为空\"}";
	        	}
	        	if(userid.isEmpty())
	        	{
	        		return "{\"result\": \"ERROR\", \"msg\":\"userid不能为空\"}";
	        	}
	            return friendproducer.send("FriendlyInfor","friend", userid, msg, Integer.parseInt(userid));
	        } catch (InterruptedException e) {
	            e.printStackTrace();
	        } catch (RemotingException e) {
	            e.printStackTrace();
	        } catch (MQClientException e) {
	            e.printStackTrace();
	        } catch (MQBrokerException e) {
	            e.printStackTrace();
	        } catch (UnsupportedEncodingException e) {
	            e.printStackTrace();
	        }
	        return "ERROR";
	    }
	    @RequestMapping("/PushLogisticsMsg")
	    public String PushLogisticsMsg(String userid, String msg){
	        try {
	        	System.out.println("短信内容：" + msg);
	        	if(msg.isEmpty())
	        	{
	        		return "{\"result\": \"ERROR\", \"msg\":\"消息内容不能为空\"}";
	        	}
	        	if(userid.isEmpty())
	        	{
	        		return "{\"result\": \"ERROR\", \"msg\":\"userid不能为空\"}";
	        	}
	            return logisticsproducer.send("LogisticsInfor","Logistics", userid, msg, Integer.parseInt(userid));
	        } catch (InterruptedException e) {
	            e.printStackTrace();
	        } catch (RemotingException e) {
	            e.printStackTrace();
	        } catch (MQClientException e) {
	            e.printStackTrace();
	        } catch (MQBrokerException e) {
	            e.printStackTrace();
	        } catch (UnsupportedEncodingException e) {
	            e.printStackTrace();
	        }
	        return "ERROR";
	    }

	    @RequestMapping("/PullFriendMsg")
	    public String PullFriendMsg(String userid, String start, String end){
	        try {
	        	if(start.isEmpty())
	        	{
	        		return "{\"result\": \"ERROR\", \"msg\":\"start不能为空\"}";
	        	}
	        	if(userid.isEmpty())
	        	{
	        		return "{\"result\": \"ERROR\", \"msg\":\"userid不能为空\"}";
	        	}
	        	if(end.isEmpty())
	        	{
	        		return "{\"result\": \"ERROR\", \"msg\":\"end不能为空\"}";
	        	}

	           // return friendconsumer.viewMessage(msgid);

	        	return friendconsumer.pull("FriendlyInfor", userid, start, end);
	        } catch (InterruptedException e) {
	            e.printStackTrace();
	        } catch (RemotingException e) {
	            e.printStackTrace();
	        } catch (MQClientException e) {
	            e.printStackTrace();
	        } catch (MQBrokerException e) {
	            e.printStackTrace();
	        } catch (UnsupportedEncodingException e) {
	            e.printStackTrace();
	        }
	        return "ERROR";
	    }
	    @RequestMapping("/PullLogisticsMsg")
	    public String PullLogisticsMsg(String userid, String start, String end){
	        try {
	        	if(start.isEmpty())
	        	{
	        		return "{\"result\": \"ERROR\", \"msg\":\"start不能为空\"}";
	        	}
	        	if(userid.isEmpty())
	        	{
	        		return "{\"result\": \"ERROR\", \"msg\":\"userid不能为空\"}";
	        	}
	        	if(end.isEmpty())
	        	{
	        		return "{\"result\": \"ERROR\", \"msg\":\"end不能为空\"}";
	        	}
	            return logisticsconsumer.pull("FriendlyInfor", userid, start, end);
	        } catch (InterruptedException e) {
	            e.printStackTrace();
	        } catch (RemotingException e) {
	            e.printStackTrace();
	        } catch (MQClientException e) {
	            e.printStackTrace();
	        } catch (MQBrokerException e) {
	            e.printStackTrace();
	        } catch (UnsupportedEncodingException e) {
	            e.printStackTrace();
	        }
	        return "ERROR";
	    }
	    @RequestMapping("/PullnotifyMsg")
	    public String PullnotifyMsg(String userid, String offset){
	        try {
	        	if(userid.isEmpty())
	        	{
	        		return "{\"result\": \"ERROR\", \"msg\":\"userid不能为空\"}";
	        	}
	        	if(offset.isEmpty())
	        	{
	        		return "{\"result\": \"ERROR\", \"msg\":\"end不能为空\"}";
	        	}
	            return notifyconsumer.pull("notifyinfor", userid, offset);
	        } catch (InterruptedException e) {
	            e.printStackTrace();
	        } catch (RemotingException e) {
	            e.printStackTrace();
	        } catch (MQClientException e) {
	            e.printStackTrace();
	        } catch (MQBrokerException e) {
	            e.printStackTrace();
	        } catch (UnsupportedEncodingException e) {
	            e.printStackTrace();
	        }
	        return "ERROR";
	    }
	    @RequestMapping("/OffsetMsgId")
	    public String pushMsgsendfrommsgid(String msgid) throws MQClientException, RemotingException, MQBrokerException, InterruptedException{

	    	if(msgid.isEmpty())
	    	{
	    		return "错误msgid";
	    	}
	        DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("pullConsumer");

	        consumer.setNamesrvAddr(namesrvAddr);
	        consumer.start();

	        System.out.printf("msgid：" + msgid + "\n");



	        MessageExt msg = consumer.viewMessage(msgid);

	        consumer.shutdown();


	        return "{\"result\": \"OK\", \"msg\":\"+new String(msg.getBody()) +\"}";
	    }
}
